package cn.juc;

import javafx.concurrent.Worker;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 *
 */
@Slf4j
public class PoolTest {
    public static void main(String[] args) {
        ThreadPool<Thread> pool=new ThreadPool<>(2,1000,TimeUnit.MILLISECONDS,10,(queue,task)->{
            //queue.offer(task,1500,TimeUnit.MILLISECONDS);
            queue.put(task);
        });
        for (int i = 0; i <5 ; i++) {
            pool.execute(()->{});
        }
    }
}
@Slf4j
class BlockingQueue<T> {
    //任务队列
    private Deque<T> tDeque=new ArrayDeque();
    //锁
    private ReentrantLock lock=new ReentrantLock();
    //生产者条件变量
    private Condition fullWaitSet=lock.newCondition();
    //消费者条件变量
    private Condition emptyWaitSet=lock.newCondition();
    //pool容量
    private int capcity;
    private RejectPolicy<T> rejectPolicy;
    public BlockingQueue(int queMax) {
        this.capcity=queMax;
    }

    //阻塞获取
    public T pool(long timeout, TimeUnit u){
        lock.lock();
        try {
            long nanos = u.toNanos(timeout);
            while (true) {
                if (tDeque.isEmpty()) { //空的无法获取 就阻塞
                    try {
                        if (nanos<=0){
                            return null;
                        }
                        //剩余时间
                        nanos= emptyWaitSet.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return tDeque.removeFirst();//给与第一个
            }
        }finally {
            lock.unlock();//解锁
        }
    }
    public T take(){
        //log.info("wo no no no no");
        lock.lock();
        try {
        while (true) {
            if (tDeque.isEmpty()) { //空的无法获取 就阻塞
                log.info("空队列的无法获取任务");
                try {
                    emptyWaitSet.await();
              //      log.info("空队列的无法获取任务");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
             log.info("给与第一个任务");
            T t = tDeque.removeFirst();
            fullWaitSet.signal();
            return t;//给与第一个
        }
        }finally {
            lock.unlock();//解锁
        }
    }
    //生产
    public void put(T o){
         lock.lock();
         try {
             if (tDeque.size()==capcity){ //容量满了 停止生产
                 try {
                     log.info("容量满了 停止生产");
                     fullWaitSet.await();
                 } catch (InterruptedException e) {
                     e.printStackTrace();
                 }
             }
             else {
                 tDeque.addLast(o);
                 log.info("队列进入一个任务");
                 emptyWaitSet.signal();//唤醒停止使用的线程
             }
         }
         finally {
             lock.unlock();
         }
    }
    //生产增强
    public boolean offer(T o,long timeout,TimeUnit unit){
        lock.lock();
        try {
            long nanos = unit.toNanos(timeout);
            log.info("我在执行===================");
            while (tDeque.size()==capcity){ //容量满了 停止生产
                try {
                  //  log.info("容量满了 停止生产");
                    if (nanos<=0) {
                        log.info("超时生产");
                        return false;
                    }
                    log.info("容量满了 停止生产");
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            tDeque.addLast(o);
            log.info("队列进入一个任务");
            emptyWaitSet.signal();//唤醒停止使用的线程
            return true;
        }
        finally {
            lock.unlock();
        }
    }

    public  void tryput(RejectPolicy<T> rejectPolicy, T task) {
       lock.lock();
       try {
           if (tDeque.size()==capcity){ //容量满了 停止生产
               rejectPolicy.reject(this,task);
           }
           else {
               tDeque.addLast(task);
               log.info("队列进入一个任务");
               emptyWaitSet.signal();//唤醒停止使用的线程
           }
       }
       finally {
           lock.unlock();
       }
    }
}
//拒绝策略
@FunctionalInterface
interface RejectPolicy<T>{
    void reject(BlockingQueue<T> queue,T task );
}
@Slf4j
class ThreadPool<T> {
    private BlockingQueue queue; //任务队列
    private HashSet<Thread> threads=new HashSet<>();//线程集合
    private Integer coreSize;//核心线程程数
    private long timeout;
    private TimeUnit timeUnit;
    private RejectPolicy<T> rejectPolicy;
    public ThreadPool(Integer coreSize, long timeout, TimeUnit timeUnit,int queMax,RejectPolicy<T> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.queue=new BlockingQueue(queMax);
        this.rejectPolicy=rejectPolicy;
    }

    class Worker extends Thread{
        private Runnable task;
        public Worker(Runnable runnable){
            this.task=runnable;
        }

        @Override
        public void run() {
            super.run();
            //传入了线程 或者可以任务队列不为空
            //log.info("死循环开始");
            //System.out.println(queue.take());
            while (task!=null||(task= (Runnable) queue.take())!=null){
               // log.info(String.valueOf(task.toString()==null));
                try {
                    task.run();
                    Thread.sleep(2000);
                    log.info("任务执行中");
                }catch (Exception e){
                }
                finally {
                    task=null;//结束任务
                }
            }
            synchronized (threads){
                threads.remove(this);
                log.info("执行完了离开队列");
            }
        }
    }
    //任务执行
    public void execute(Runnable task){
        //没有超过线程数 执行
        synchronized (threads) {
            if (threads.size() <= coreSize) {
                Worker worker = new Worker(task);
                log.info("正在成功使用");
                threads.add(worker);
                worker.start();//启动线程
            }
            //超过线程数
            else {
                //queue.put(task);//加入任务队列
                //log.info("加入任务");
                queue.tryput(rejectPolicy,task);
            }
        }
    }
}